package producer

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	uuid "github.com/satori/go.uuid"
	"github.com/zeromicro/go-zero/core/logx"
	"go-zero-project/common/excption"
	tools "go-zero-project/common/utils"
	"go-zero-project/services/user/api/constans"
	"go-zero-project/services/user/api/internal/config"
	"sync"
)

var transMap = sync.Map{}

type MqTransBO struct {
	// 发送消息Topic
	topic string
	// 发生消息体
	msg *primitive.Message
	// 本地事务执行方法
	localTransFun func(msg *primitive.Message) primitive.LocalTransactionState
	// 会查事务执行方法
	checkLocalFun func(msg *primitive.MessageExt) primitive.LocalTransactionState
}

func NewMqTransBO(topic string, msg *primitive.Message,
	localTransFun func(msg *primitive.Message) primitive.LocalTransactionState,
	checkLocalFun func(msg *primitive.MessageExt) primitive.LocalTransactionState) *MqTransBO {
	return &MqTransBO{topic: topic, msg: msg, localTransFun: localTransFun, checkLocalFun: checkLocalFun}
}

type TransListener struct {
}

func NewTransListener() *TransListener {
	return &TransListener{}
}

func (dl *TransListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	// 从map中那拿到 执行本地方法的具体函数
	trans, ok := transMap.Load(msg.Topic)
	if !ok {
		logx.Errorf("not found topic:%s  ExecuteLocalTransaction func", msg.Topic)
		return primitive.RollbackMessageState
	}
	bo := trans.(*MqTransBO)
	// 设置默认状态码
	transFuncState := primitive.UnknowState
	// 执行具体事务消息的函数
	transFuncState = bo.localTransFun(msg)
	// 处理下异常
	defer (func(state primitive.LocalTransactionState) {
		if err := recover(); err != nil {
			logx.Errorf("%s ExecuteLocalTransaction error:%s", msg.Topic, err)
			transFuncState = primitive.RollbackMessageState
		}
	})(transFuncState)
	// 返回结果
	return transFuncState

}

func (dl *TransListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	// 从map中拿到会查 执行回查方法逻辑
	trans, ok := transMap.Load(msg.Topic)
	if !ok {
		logx.Errorf("not found topic:%s  CheckLocalTransaction func", msg.Topic)
		return primitive.RollbackMessageState
	}
	bo := trans.(*MqTransBO)
	// 设置默认状态码
	transFuncState := primitive.RollbackMessageState
	// 执行回查方法
	transFuncState = bo.checkLocalFun(msg)
	// 处理下异常
	defer (func(state primitive.LocalTransactionState) {
		if err := recover(); err != nil {
			logx.Errorf("%s ExecuteLocalTransaction error:%s", msg.Topic, err)
		}
	})(transFuncState)

	return transFuncState
}

// SendTransMsg 发送事务消息封装
func (bo *MqTransBO) SendTransMsg(ctx context.Context, producer rocketmq.TransactionProducer) error {
	// 判断在 transMap 是否存在则返回 不存在这村塾
	actual, loaded := transMap.LoadOrStore(bo.topic, bo)
	if loaded {
		logx.Infof("the transMap not exists loaded: %s", tools.BeanToJson(actual))
	}
	transaction, sendTransErr := producer.SendMessageInTransaction(ctx, bo.msg)
	if sendTransErr != nil {
		logx.Errorf("sendTrans %s Err", sendTransErr.Error())
		return sendTransErr
	}
	logx.Infof("send sendTrans topic:%s msg:%s ,result:%s success", bo.topic, string(bo.msg.Body),
		tools.BeanToJson(transaction))
	return nil
}

func InitTransProducer(config *config.MqConfig) (rocketmq.TransactionProducer, error) {
	list := config.TopicList
	if len(list) < 1 {
		logx.Errorf("not setting topic info, check conf...")
		return nil, excption.NewErrCode(constans.MqSkipLoadErr)
	}
	if !containsTransTopic(list) {
		logx.Errorf("topic all trans Topic,not init trans producer ...")
		return nil, excption.NewErrCode(constans.MqSkipLoadErr)
	}
	// 地址
	namesrvs := []string{config.Namesrv}
	// 生产者组名称
	var groupName = config.GroupName
	// 设置消息轨迹
	traceCfg := &primitive.TraceConfig{
		Access:   primitive.Local,
		Resolver: primitive.NewPassthroughResolver(namesrvs),
	}
	// 高并发保证name唯一 多个producer 需要设置实例Id
	instanceId := uuid.NewV4()
	p, _ := rocketmq.NewTransactionProducer(
		NewTransListener(),
		producer.WithNsResolver(primitive.NewPassthroughResolver(namesrvs)),
		producer.WithRetry(1),
		producer.WithGroupName(groupName),
		producer.WithTrace(traceCfg),
		producer.WithInstanceName(fmt.Sprintf("%d", instanceId)),
	)

	err := p.Start()
	if err != nil {
		fmt.Printf("start producer error: %s\n", err.Error())
		return nil, err
	}
	logx.Infof("init transProducer success !")
	return p, err
}

func containsTransTopic(list []config.TopicConfig) bool {
	for _, v := range list {
		if v.TopicType == 4 {
			return true
		}
	}
	return false
}
